package com.boot.study.sink

import java.util

import com.boot.study.api.SensorReading
import org.apache.flink.api.common.functions.RuntimeContext
import org.apache.flink.streaming.api.scala.{StreamExecutionEnvironment, _}
import org.apache.flink.streaming.connectors.elasticsearch.{ElasticsearchSinkFunction, RequestIndexer}
import org.apache.flink.streaming.connectors.elasticsearch6.ElasticsearchSink
import org.apache.http.HttpHost
import org.elasticsearch.action.index.IndexRequest
import org.elasticsearch.client.Requests


object EsSinkTest {
  def main(args: Array[String]): Unit = {
    // 创建执行环境
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    //    // 读取文件
    val inputSteam: DataStream[String] = env.readTextFile("D:\\WorkSpace\\idea\\Flink\\src\\main\\resources\\sensor.txt")
    val dataStream: DataStream[SensorReading] = inputSteam.map(data => {
      val arr: Array[String] = data.split(",")
      SensorReading(arr(0), arr(1).toLong, arr(2).toDouble)
    })

    dataStream.print("es sink")

    // 定义httpHost
    val httpHosts: util.List[HttpHost] = new util.ArrayList[HttpHost]()
    httpHosts.add(new HttpHost("127.0.0.1", 9200))

    // 自定义写入es的EsSinkFunction
    val myEsSinkFunction = new ElasticsearchSinkFunction[SensorReading] {
      override def process(element: SensorReading, ctx: RuntimeContext, indexer: RequestIndexer): Unit = {
        // 包装一个map作为data source
        val dataSource: util.Map[String, String] = new util.HashMap[String, String]()
        dataSource.put("id", element.id)
        dataSource.put("timeStamp", element.timeStamp.toString)
        dataSource.put("temperature", element.temperature.toString)

        // 创建index request 用于发送http请求
        val indexRequest: IndexRequest = Requests.indexRequest()
          .index("sensor")
          .`type`("redingdata")
          .source(dataSource)

        // 用indexer发送数据
        indexer.add(indexRequest)
      }
    }

    dataStream.addSink(new ElasticsearchSink.Builder[SensorReading](
      httpHosts,
      myEsSinkFunction
    ).build())

    // es 浏览器测试地址 http://127.0.0.1:9200/sensor/_search?pretty

    // 执行
    env.execute("es sink test")
  }
}
